package io.github.springstudent.dekstop.client.compress;

import io.github.springstudent.dekstop.client.bean.Capture;
import io.github.springstudent.dekstop.client.bean.Listeners;
import io.github.springstudent.dekstop.client.capture.CaptureEngineListener;
import io.github.springstudent.dekstop.client.concurrent.DefaultThreadFactoryEx;
import io.github.springstudent.dekstop.client.concurrent.Executable;
import io.github.springstudent.dekstop.client.squeeze.Compressor;
import io.github.springstudent.dekstop.client.squeeze.NullTileCache;
import io.github.springstudent.dekstop.client.squeeze.RegularTileCache;
import io.github.springstudent.dekstop.client.squeeze.TileCache;
import io.github.springstudent.dekstop.common.bean.CompressionMethod;
import io.github.springstudent.dekstop.common.bean.MemByteBuffer;
import io.github.springstudent.dekstop.common.configuration.CompressorEngineConfiguration;
import io.github.springstudent.dekstop.common.configuration.ReConfigurable;
import io.github.springstudent.dekstop.common.log.Log;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class CompressorEngine implements ReConfigurable<CompressorEngineConfiguration>, CaptureEngineListener {
    private final Listeners<CompressorEngineListener> listeners = new Listeners<>();

    private ThreadPoolExecutor executor;

    private TileCache cache;

    private final Object reconfigurationLOCK = new Object();

    private CompressorEngineConfiguration configuration;

    private boolean reconfigured;

    @Override
    public void configure(CompressorEngineConfiguration configuration) {
        synchronized (reconfigurationLOCK) {
            this.configuration = configuration;
            this.reconfigured = true;
        }
    }

    @Override
    public void reconfigure(CompressorEngineConfiguration configuration) {
        configure(configuration);
    }

    public void addListener(CompressorEngineListener listener) {
        listeners.add(listener);
    }

    public void start(int queueSize) {
        // THREAD = 1
        //
        // The parallel processing is within the compressor itself - here we
        // want
        // to ensure a certain order of processing - if need more than one
        // thread
        // then have a look how the compressed data are sent over the network
        // (!)

        // QUEUESIZE = 1
        //
        // Do we need more than one here ?
        //
        // - queue full because we could not compress the last capture when a
        // new
        // one is available => too many captures (!)
        //
        // - we could not send the last compressed capture over the network as
        // the
        // network queue is full => too many capture (!)
        Log.debug("CompressorEngine start");

        executor = new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new ArrayBlockingQueue<>(queueSize));

        executor.setThreadFactory(new DefaultThreadFactoryEx("CompressorEngine"));

        executor.setRejectedExecutionHandler((runnable, poolExecutor) -> {
            if (!poolExecutor.isShutdown()) {
                final List<Runnable> pendings = new ArrayList<>();
                // pendings : oldest first (!)
                poolExecutor.getQueue().drainTo(pendings);
                final MyExecutable newer = (MyExecutable) runnable;
                if (!pendings.isEmpty()) {
                    final Capture[] cpendings = new Capture[pendings.size()];
                    for (int pos = 0, idx = pendings.size() - 1; idx > -1; pos++, idx--) {
                        cpendings[pos] = ((MyExecutable) pendings.get(idx)).getCapture();
                    }
                    newer.getCapture().mergeDirtyTiles(cpendings);
                }
                poolExecutor.execute(newer);
            }
        });
    }

    /**
     * Must not block.
     * <p/>
     * Each capture is posted in the right order (according to its capture-id)
     * from a SINGLE thread. Have a look to the rejection execution handler =>
     * between the poll() and re-execute() another thread must not execute
     * during the rejection processing to keep the right order of the captures
     * in the queue.
     * <p/>
     * Note that the current implementation should never block and never throw
     * any rejected exception.
     */
    @Override
    public void onCaptured(Capture capture) {
        executor.execute(new MyExecutable(capture));
    }

    @Override
    public void onRawCaptured(int id, byte[] grays) {
        // debugging purpose (!)
    }

    public void stop() {
        Log.debug("CompressorEngine stop");
        if (executor == null) {
            return;
        }
        executor.shutdown();
    }

    private class MyExecutable extends Executable {
        private final Capture capture;

        MyExecutable(Capture capture) {
            super(executor);
            this.capture = capture;
        }

        @Override
        protected void execute() throws IOException {
            final CompressorEngineConfiguration xconfiguration;
            final boolean xreconfigured;
            synchronized (reconfigurationLOCK) {
                xconfiguration = configuration;
                xreconfigured = reconfigured;
                if (reconfigured) {
                    cache = xconfiguration.useCache() ? new RegularTileCache(xconfiguration.getCacheMaxSize(), xconfiguration.getCachePurgeSize())
                            : new NullTileCache();
                    reconfigured = false;
                    Log.info("Compressor engine has been reconfigured [tile:" + capture.getId() + "] " + xconfiguration);
                }
            }
            final Compressor compressor = Compressor.get(xconfiguration.getMethod());
            final MemByteBuffer compressed = compressor.compress(cache, capture);

            // Possibly blocking - no problem as we'll replace (and merge) in our queue
            // the oldest capture (if any) until we can compress it and send it to the next
            // stage of processing.
            if (!xreconfigured) {
                fireOnCompressed(capture.getId(), compressor.getMethod(), null, compressed);
            } else {
                // we have to send the whole configuration => de-compressor synchronization (!)
                fireOnCompressed(capture.getId(), compressor.getMethod(), xconfiguration, compressed);
            }
            cache.onCaptureProcessed();
        }

        private void fireOnCompressed(int captureId, CompressionMethod compressionMethod, CompressorEngineConfiguration compressionConfiguration,
                                      MemByteBuffer compressed) {
            listeners.getListeners().forEach(listener -> listener.onCompressed(captureId, compressionMethod, compressionConfiguration, compressed));
        }

        Capture getCapture() {
            return capture;
        }
    }

}
